package com.alibaba.csp.sentinel.dashboard.repository.metric;

import com.alibaba.csp.sentinel.dashboard.datasource.entity.InfluxdbMetricEntity;
import com.alibaba.csp.sentinel.dashboard.datasource.entity.MetricEntity;
import com.alibaba.csp.sentinel.util.StringUtil;
import lombok.extern.slf4j.Slf4j;
import org.influxdb.InfluxDB;
import org.influxdb.dto.BatchPoints;
import org.influxdb.dto.Point;
import org.influxdb.dto.Query;
import org.influxdb.dto.QueryResult;
import org.influxdb.impl.InfluxDBResultMapper;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

@Slf4j
@Component
public class InInfluxdbMetricsRepository implements MetricsRepository<MetricEntity> {
    private static final String MEASUREMENT_NAME = "sentinel_metric";

    @Autowired
    public InfluxDB influxDB;

    @Override
    public synchronized void save(MetricEntity metric) {
        influxDB.write(createPoint(metric, System.currentTimeMillis()));
    }

    @Override
    public synchronized void saveAll(Iterable<MetricEntity> metrics) {
        if (metrics == null) {
            return;
        }
        BatchPoints batchPoints = BatchPoints.builder().consistency(InfluxDB.ConsistencyLevel.ALL).build();

        long currentTimeMillis = System.currentTimeMillis();
        AtomicInteger counter = new AtomicInteger();
        metrics.forEach(metric -> {
            long timeMillis = currentTimeMillis + counter.incrementAndGet();
            batchPoints.point(createPoint(metric, timeMillis));
        });
        influxDB.write(batchPoints);
    }

    private Point createPoint(MetricEntity metric, long timeMillis) {
        // 非网关应用则资源名前缀加上应用名
        String resource = metric.getResource();
        if (!metric.getApp().contains("gateway")) {
            resource = "/" + metric.getApp() + resource;
        }
        return Point.measurement(MEASUREMENT_NAME)
                .time(timeMillis, TimeUnit.MILLISECONDS)
                .tag(InfluxdbMetricEntity.APP, metric.getApp())
                .tag(InfluxdbMetricEntity.INSTANCE, metric.getInstance())
                .tag(InfluxdbMetricEntity.RESOURCE, resource)
                .addField(InfluxdbMetricEntity.GMT_CREATE, metric.getGmtCreate().getTime())
                .addField(InfluxdbMetricEntity.GMT_MODIFIED, metric.getGmtModified().getTime())
                .addField(InfluxdbMetricEntity.TIMESTAMP, metric.getTimestamp().getTime())
                .addField(InfluxdbMetricEntity.PASS_QPS, metric.getPassQps())
                .addField(InfluxdbMetricEntity.SUCCESS_QPS, metric.getSuccessQps())
                .addField(InfluxdbMetricEntity.BLOCK_QPS, metric.getBlockQps())
                .addField(InfluxdbMetricEntity.EXCEPTION_QPS, metric.getExceptionQps())
                .addField(InfluxdbMetricEntity.RT, metric.getRt())
                .addField(InfluxdbMetricEntity.COUNT, metric.getCount())
                .addField(InfluxdbMetricEntity.RESOURCE_CODE, metric.getResourceCode())
                .build();
    }

    @Override
    public synchronized List<MetricEntity> queryByAppAndResourceBetween(String app, String resource, long startTime, long endTime) {
        List<MetricEntity> results = new ArrayList<>();
        if (StringUtil.isBlank(app)) {
            return results;
        }
        String command = String.format("SELECT * FROM %s WHERE app='%s' AND resource = '%s' AND gmtCreate>%d AND gmtCreate<%d",
                MEASUREMENT_NAME, app, resource, startTime, endTime);
        Query query = new Query(command);
        QueryResult queryResult = influxDB.query(query);
        InfluxDBResultMapper resultMapper = new InfluxDBResultMapper();
        List<InfluxdbMetricEntity> influxResults = resultMapper.toPOJO(queryResult, InfluxdbMetricEntity.class);
        try {
            influxResults.forEach(entity -> {
                MetricEntity metric = new MetricEntity();
                BeanUtils.copyProperties(entity, metric);
                metric.setTimestamp(new Date(entity.getTimestamp()));
                metric.setGmtCreate(new Date(entity.getGmtCreate()));
                metric.setGmtModified(new Date(entity.getGmtModified()));
                results.add(metric);
            });
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
        return results;
    }

    @Override
    public synchronized List<String> listResourcesOfApp(String app, long startTime, long endTime) {
        List<String> results = new ArrayList<>();
        if (StringUtil.isBlank(app)) {
            return results;
        }
        String command = String.format("SELECT * FROM %s WHERE app='%s' AND gmtCreate>%d AND gmtCreate<%d",
                MEASUREMENT_NAME, app, startTime, endTime);
        Query query = new Query(command);
        QueryResult queryResult = influxDB.query(query);
        InfluxDBResultMapper resultMapper = new InfluxDBResultMapper();
        List<InfluxdbMetricEntity> influxResults = resultMapper.toPOJO(queryResult, InfluxdbMetricEntity.class);
        try {
            if (CollectionUtils.isEmpty(influxResults)) {
                return results;
            }
            Map<String, InfluxdbMetricEntity> resourceCount = new HashMap<>(32);
            for (InfluxdbMetricEntity metricEntity : influxResults) {
                String resource = metricEntity.getResource();
                if (resourceCount.containsKey(resource)) {
                    InfluxdbMetricEntity oldEntity = resourceCount.get(resource);
                    oldEntity.addPassQps(metricEntity.getPassQps());
                    oldEntity.addRtAndSuccessQps(metricEntity.getRt(), metricEntity.getSuccessQps());
                    oldEntity.addBlockQps(metricEntity.getBlockQps());
                    oldEntity.addExceptionQps(metricEntity.getExceptionQps());
                    oldEntity.addCount(1);
                } else {
                    InfluxdbMetricEntity influxdbMetricEntity = new InfluxdbMetricEntity();
                    BeanUtils.copyProperties(metricEntity, influxdbMetricEntity);
                    influxdbMetricEntity.setTime(null);
                    resourceCount.put(resource, influxdbMetricEntity);
                }
            }
            // 排序
            results = resourceCount.entrySet()
                    .stream()
                    .sorted((o1, o2) -> {
                        InfluxdbMetricEntity e1 = o1.getValue();
                        InfluxdbMetricEntity e2 = o2.getValue();
                        int t = e2.getBlockQps().compareTo(e1.getBlockQps());
                        if (t != 0) {
                            return t;
                        }
                        return e2.getPassQps().compareTo(e1.getPassQps());
                    })
                    .map(Map.Entry::getKey)
                    .collect(Collectors.toList());
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
        return results;
    }
}
